package com.sumbo.config.reader;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.sumbo.config.constant.RocketMqConstants;
import com.sumbo.config.pipeline.PipelineDispatch;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * \* User: MeiZhongHao
 * \* Date: 2019-06-14
 * \* Time: 11:39
 * \* Description:
 * \
 */
public class RocketMqReader extends Reader {

    public RocketMqReader() {
    }

    public static RocketMqReader instance = new RocketMqReader();

    public static Map<String, DefaultMQPushConsumer> consumerMap = new HashMap<>();

    public static ExecutorService executor;

    public void init(Map<String, Properties> temp) {
        if (temp == null || temp.size() <= 0) {
            return;
        }
        for (String key : temp.keySet()) {
            String consumeGroupName = temp.get(key).getProperty(RocketMqConstants.consumeGroup);
            String nameServerAddr = temp.get(key).getProperty(RocketMqConstants.namesrvAddr);
            String instanceName = temp.get(key).getProperty(RocketMqConstants.instanceName);
            String messageModel = temp.get(key).getProperty(RocketMqConstants.messageModel);
            String topic = temp.get(key).getProperty(RocketMqConstants.topic);
            String tag = temp.get(key).getProperty(RocketMqConstants.tag);
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumeGroupName);
            consumer.setNamesrvAddr(nameServerAddr);
            if (StringUtils.isEmpty(instanceName)) {
                instanceName = UUID.randomUUID().toString();
            }
            consumer.setInstanceName(instanceName);
            if ("B".equals(messageModel)) {
                consumer.setMessageModel(MessageModel.BROADCASTING);
            } else if ("C".equals(messageModel)) {
                consumer.setMessageModel(MessageModel.CLUSTERING);
            }
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            try {
                consumer.subscribe(topic, tag);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    for (MessageExt message : msgs) {
                        String msg = new String(message.getBody());
                        if (isValidJson(msg)) {
                            PipelineDispatch.dispatch(key, msg);
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
        }
        executor = Executors.newFixedThreadPool(consumerMap.size());
    }

    @Override
    public void open() {
        for (String name : consumerMap.keySet()) {
            executor.submit(() -> read(name));
        }
    }

    @Override
    public void close() {

    }

    public void read(String name) {
        try {
            consumerMap.get(name).start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    private static boolean isValidJson(String content) {
        try {
            JSONObject jsonStr = JSONObject.parseObject(content);
            return true;
        } catch (Exception e) {
            return false;
        }
    }
}